package com.yangye.niodemo.multiplexing;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * @author yangye
 * @date 2020/12/22
 */
public class SocketMultiplexingSingleThread {

	private ServerSocketChannel server = null;

	private Selector selector = null;

	int port = 9090;

	public void initServer() {
		try {
			server = ServerSocketChannel.open();
			server.configureBlocking(false);
			server.bind(new InetSocketAddress(port));

			// epoll:epoll_create
			selector = Selector.open();

			// select,poll:jvm开辟一个数组，fd放进去
			// epoll:epoll_ctl(fd3,ADD,fd4,EPOLLIN)
			server.register(selector, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void start() {
		initServer();
		System.out.println("服务器启动了...");
		try {
			while (true) {
				Set<SelectionKey> keys = selector.keys();
				System.out.println("key size " + keys.size());
				// 1.调用多路复用器
				/**
				 * select,poll:调用内核的select(fds)方法
				 * epoll:调用内核的epoll_wait
				 */
				System.out.println("准备阻塞");
				while (selector.select(500) > 0) {
					System.out.println("获取到fd");
					// 返回的有状态的fd集合
					Set<SelectionKey> selectionKeys = selector.selectedKeys();
					Iterator<SelectionKey> iterator = selectionKeys.iterator();
					// 处理fd的R/W
					while (iterator.hasNext()) {
						SelectionKey key = iterator.next();
						iterator.remove();
						if (key.isAcceptable()) {
							handleAccept(key);
						} else if (key.isReadable()) {
							handleRead(key);
						}
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private void handleRead(SelectionKey key) {
		SocketChannel client = (SocketChannel) key.channel();
		ByteBuffer buffer = (ByteBuffer) key.attachment();
		buffer.clear();

		int read = 0;
		try {
			while (true) {
				read = client.read(buffer);
				if (read > 0) {
					buffer.flip();
					while (buffer.hasRemaining()) {
						client.write(buffer);
					}
					buffer.clear();
				} else if (read == 0) {
					break;
				} else {
					client.close();
					break;
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	private void handleAccept(SelectionKey key) {
		try {
			ServerSocketChannel channel = (ServerSocketChannel) key.channel();
			SocketChannel client = channel.accept();
			client.configureBlocking(false);

			ByteBuffer buffer = ByteBuffer.allocate(8192);
			/**
			 * select,poll:jvm开辟一个数组，fd放进去
			 * epoll:epoll_ctl(fd3,ADD,fd4,EPOLLIN)
			 */
			client.register(selector, SelectionKey.OP_READ, buffer);
			System.out.println("--------------------");
			System.out.println("新客户端连接成功：" + client.getRemoteAddress());
			System.out.println("--------------------");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		SocketMultiplexingSingleThread server = new SocketMultiplexingSingleThread();
		server.start();
	}
}
